package com.bw.app.dwm;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.app.func.DimAsyncFunctionNew1;
import com.bw.bean.OrderDetail;
import com.bw.bean.OrderInfo;
import com.bw.bean.OrderWide;
import com.bw.common.GmallConfig;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @ProjectName: BigData
 * @Package: com.bw.app.dwm
 * @ClassName: OrderWideApp
 * @Author: Gy
 * @Description: 合并订单宽表
 * @Date: 2021/11/11 15:39
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        //todo 1.基本环境准备
        //1.1 准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2设置并行度
        env.setParallelism(4);

        //todo 2.从kafka的DWD层读取订单和订单明细数据
        //2.1 声明相关的主题以及消费者组
        String orderInfoSourceTopic = "dwd_order_info";
        String orderDetailSourceTopic = "dwd_order_detail";
        String orderWideSinkTopic = "dwm_order_wide";
        String groupId = "order_wide_group";

        //2.2 读取订单主题数据
        FlinkKafkaConsumer<String> orderInfoSource = MyKafkaUtil.getKafkaSource(orderInfoSourceTopic, groupId);
        DataStreamSource<String> orderInfoJsonStrDs = env.addSource(orderInfoSource);

        //2.3 读取订单明细数据
        FlinkKafkaConsumer<String> orderDetailSource = MyKafkaUtil.getKafkaSource(orderDetailSourceTopic,groupId);
        DataStreamSource<String> orderDetailJsonStrDs = env.addSource(orderDetailSource);

        //todo 3.对读取的数据进行结构的转换      jsonString ——> OrderInfo|OrderDetail
        //3.1 转换订单数据结构
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = orderInfoJsonStrDs.map(new MapFunction<String, OrderInfo>() {
            @Override
            public OrderInfo map(String s) throws Exception {
                OrderInfo orderInfo = JSON.parseObject(s, OrderInfo.class);
                return orderInfo;
            }
        });

        //3.2 转换订单明细数据
        SingleOutputStreamOperator<OrderDetail> orderDetailDS = orderDetailJsonStrDs.map(new MapFunction<String, OrderDetail>() {
            @Override
            public OrderDetail map(String s) throws Exception {
                OrderDetail orderDetail = JSON.parseObject(s, OrderDetail.class);
                return orderDetail;
            }
        });

        //todo 4.指定事件时间字段
        //4.1 订单指定事件时间字段
        SingleOutputStreamOperator<OrderInfo> orderInfoWithTsDS = orderInfoDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                            @Override
                            public long extractTimestamp(OrderInfo t, long l) {
                                return t.getCreate_ts();
                            }
                        })
        );

        //4.2 订单明细指定事件时间字段
        SingleOutputStreamOperator<OrderDetail> orderDetailWithTsDS = orderDetailDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
                            @Override
                            public long extractTimestamp(OrderDetail orderDetail, long l) {
                                return orderDetail.getCreate_ts();
                            }
                        })
        );

        //todo 5.按照订单id进行分组     指定关联的key
        KeyedStream<OrderInfo, Long> orderInfoKeyedDS = orderInfoWithTsDS.keyBy(OrderInfo::getId);
        KeyedStream<OrderDetail, Long> orderDetailKeyedDS = orderDetailWithTsDS.keyBy(OrderDetail::getOrder_id);

        //todo 6.使用intervalJoin对订单和订单明细进行关联
        SingleOutputStreamOperator<JSONObject> orderWideDS = orderInfoKeyedDS
                .intervalJoin(orderDetailKeyedDS)
                .between(Time.milliseconds(-5), Time.milliseconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, JSONObject>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context context, Collector<JSONObject> collector) throws Exception {
                        OrderWide orderWide = new OrderWide(orderInfo, orderDetail);
                        //将合并的对象进行再一次的转化为json对象为后面的方便合并做准备
                        String jsonString = JSON.toJSONString(orderWide);
                        JSONObject jsonObject = JSON.parseObject(jsonString);
                        collector.collect(jsonObject);
                    }
                });

        orderWideDS.print("orderWide>>>>>");

        //todo 7.关联用户维度
        SingleOutputStreamOperator<JSONObject> orderWideWithUserDS = AsyncDataStream.unorderedWait(orderWideDS,
                new DimAsyncFunctionNew1(GmallConfig.DIM_USER_INFO),
                60, TimeUnit.SECONDS);
        orderWideWithUserDS.print(">>>>>");

        //todo 8.关联省市维度
        SingleOutputStreamOperator<JSONObject> orderWideWithProvice = AsyncDataStream.unorderedWait(
                orderWideWithUserDS,
                new DimAsyncFunctionNew1(GmallConfig.DIM_PROVICE),
                60, TimeUnit.SECONDS);
        //orderWideWithProvice.print(">>>>>");

        //todo 9.关联SKU商品维度
        SingleOutputStreamOperator<JSONObject> orderWideWithSku = AsyncDataStream.unorderedWait(
                orderWideWithProvice,
                new DimAsyncFunctionNew1(GmallConfig.DIM_SKU_INFO),
                60, TimeUnit.SECONDS);
        //orderWideWithSku.print(">>>>>");

        //todo 10.关联SPU商品维度
        SingleOutputStreamOperator<JSONObject> orderWideWithSpu = AsyncDataStream.unorderedWait(
                orderWideWithSku,
                new DimAsyncFunctionNew1(GmallConfig.DIM_SPU_INFO),
                60, TimeUnit.SECONDS);
        //orderWideWithSpu.print(">>>>>");

        //todo 11.关联品类维度
        SingleOutputStreamOperator<JSONObject> orderWideWithCatege3 = AsyncDataStream.unorderedWait(
                orderWideWithSpu,
                new DimAsyncFunctionNew1(GmallConfig.DIM_BASE_CATEGORY3),
                60, TimeUnit.SECONDS);
        //orderWideWithCatege3.print(">>>>>");

        //todo 12.关联品牌维度
        SingleOutputStreamOperator<JSONObject> orderWideWithTrademark = AsyncDataStream.unorderedWait(
                orderWideWithCatege3,
                new DimAsyncFunctionNew1(GmallConfig.DIM_BASE_TRADEMARK),
                60, TimeUnit.SECONDS);
        orderWideWithTrademark.print(">>>>>");

        //todo 13.将关联后的订单宽表数据写回到kafka的DWM层
        orderWideWithTrademark.map(
                orderWide -> JSON.toJSONString(orderWide)
        ).addSink(MyKafkaUtil.getKafkaSink(orderWideSinkTopic));

        env.execute();
    }
}
